/*
 * Copyright 2013-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.base.data.elasticsearch.repository.support;

import com.base.data.elasticsearch.core.operation.BaseElasticsearchOperations;
import com.base.data.elasticsearch.core.query.*;
import com.base.data.elasticsearch.core.syncmapping.BaseElasticMappingHandler;
import com.base.data.elasticsearch.repository.BaseElasticsearchRepository;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.index.query.QueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.*;
import org.springframework.util.Assert;

import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;

/**
 * Elasticsearch specific repository implementation. Likely to be used as target within
 * {@link BaseElasticsearchRepositoryFactory}
 *
 * @author magina
 */
public abstract class BaseAbstractElasticsearchRepository<T, ID extends Serializable>
		implements BaseElasticsearchRepository<T, ID> {

	static final Logger LOGGER = LoggerFactory.getLogger(BaseAbstractElasticsearchRepository.class);
	protected BaseElasticsearchOperations elasticsearchOperations;
	protected Class<T> entityClass;
	protected BaseElasticsearchEntityInformation<T, ID> entityInformation;

	public BaseAbstractElasticsearchRepository() {}

	public BaseAbstractElasticsearchRepository(BaseElasticsearchOperations elasticsearchOperations) {

		Assert.notNull(elasticsearchOperations, "ElasticsearchOperations must not be null!");

		this.setElasticsearchOperations(elasticsearchOperations);
	}

	public BaseAbstractElasticsearchRepository(BaseElasticsearchEntityInformation<T, ID> metadata,
                                               BaseElasticsearchOperations elasticsearchOperations) {
		this(elasticsearchOperations);

		Assert.notNull(metadata, "ElasticsearchEntityInformation must not be null!");

		this.entityInformation = metadata;
		setEntityClass(this.entityInformation.getJavaType());
		try {
			if (createIndexAndMapping()) {
//				createIndex();
//				putMapping();
				// fixed update mapping
				if(elasticsearchOperations.autoMapping()) {
					BaseElasticMappingHandler.initClazzStructure(getEntityClass(),elasticsearchOperations);
				}

				/**
				 * XContentBuilder xContentBuilder = XContentFactory.jsonBuilder();
				 * 				xContentBuilder.startObject();
				 * 				CreateIndexRequest indexRequest = new CreateIndexRequest("");
				 * 				indexRequest.mapping("asd",xContentBuilder);
				 * 				IndicesAdminClient client = null;
				 * 				final ActionFuture<CreateIndexResponse> createIndexResponseActionFuture = client.create(indexRequest);
				 */
				/**
				 * CreateIndexRequest indexRequest = new CreateIndexRequest("");
				 * 				indexRequest.mapping("asd", Optional.ofNullable(null));
				 * 				RestHighLevelClient1 client= null;
				 * 				final GetResponse response1 = client.performRequestAndParseEntity1(new GetRequest(), Request::get, GetResponse::fromXContent, singleton(404), new BasicHeader("asd", "asd"));
				 * 			//	indexRequest 未extend ActionRequest 7.x版本es highlevel 已经Deprecated，走这种方式好像行不通
				 * 		 //已经使用protected final <Req extends Validatable, Resp> Resp performRequestAndParseEntity(Req request, CheckedFunction<Req, Request, IOException> requestConverter, RequestOptions options, CheckedFunction<XContentParser, Resp, IOException> entityParser, Set<Integer> ignores) throws IOException {
				 * 				CreateIndexResponse response =  client.performRequestAndParseEntity1(indexRequest,
				 * 						RestHighLevelClient1::createIndex,
				 * 						 CreateIndexResponse1::fromXContent,
				 * 						singleton(404),
				 * 						new BasicHeader("asd","asd")
				 * 				);
				 */
			}
		} catch (ElasticsearchException exception) {
			LOGGER.error("failed to load elasticsearch nodes : " + exception.getDetailedMessage());
		}
	}

	private void createIndex() {
		elasticsearchOperations.createIndex(getEntityClass());
	}

	private void putMapping() {
		elasticsearchOperations.putMapping(getEntityClass());
	}

	private boolean createIndexAndMapping() {
		return elasticsearchOperations.getPersistentEntityFor(getEntityClass()).isCreateIndexAndMapping();
	}

	//done
	@Override
	public Optional<T> findById(ID id) {
		BaseGetQuery query = new BaseGetQuery();
		query.setId(stringIdRepresentation(id));
		return Optional.ofNullable(elasticsearchOperations.queryForObject(query, getEntityClass()));
	}


	//done
	@Override
	public Iterable<T> findAll() {
		int itemCount = (int) this.count();
		if (itemCount == 0) {
			return new PageImpl<>(Collections.<T> emptyList());
		}
		return this.findAll(PageRequest.of(0, Math.max(1, itemCount)));
	}
	//done
	@Override
	public Page<T> findAll(Pageable pageable) {
		BaseSearchQuery query = new BaseNativeSearchQueryBuilder().withQuery(matchAllQuery()).withPageable(pageable).build();
		return elasticsearchOperations.queryForPage(query, getEntityClass());
	}
	//done
	@Override
	public Iterable<T> findAll(Sort sort) {
		int itemCount = (int) this.count();
		if (itemCount == 0) {
			return new PageImpl<>(Collections.<T> emptyList());
		}
		BaseSearchQuery query = new BaseNativeSearchQueryBuilder().withQuery(matchAllQuery()).withPageable(PageRequest.of(0, itemCount, sort)).build();
		return elasticsearchOperations.queryForPage(query, getEntityClass());
	}
	//done
	@Override
	public Iterable<T> findAllById(Iterable<ID> ids) {
		Assert.notNull(ids, "ids can't be null.");
		BaseSearchQuery query = new BaseNativeSearchQueryBuilder().withIds(stringIdsRepresentation(ids)).build();
		return elasticsearchOperations.multiGet(query, getEntityClass());
	}
	//done
	@Override
	public long count() {
		BaseSearchQuery query = new BaseNativeSearchQueryBuilder().withQuery(matchAllQuery()).build();
		return elasticsearchOperations.count(query, getEntityClass());
	}
	//done
	@Override
	public <S extends T> S save(S entity) {
		Assert.notNull(entity, "Cannot save 'null' entity.");
		elasticsearchOperations.index(createIndexQuery(entity));
		return entity;
	}


	//done
	@Override
	public <S extends T> S index(S entity) {
		return save(entity);
	}
	//done
	@Override
	public <S extends T> Iterable<S> saveAll(Iterable<S> entities) {
		Assert.notNull(entities, "Cannot insert 'null' as a List.");
		List<BaseIndexQuery> queries = new ArrayList<>();
		LOGGER.debug("{} - start ",Thread.currentThread().getStackTrace()[1].getMethodName());
		for (S s : entities) {
			queries.add(createIndexQuery(s));
		}
		LOGGER.debug("{} - createIndexQuery end ",Thread.currentThread().getStackTrace()[1].getMethodName());
		elasticsearchOperations.bulkIndex(queries);
		return entities;
	}

	//done
	@Override
	public <S extends T> Iterable<S> saveAllAsync(Iterable<S> entities) {
		Assert.notNull(entities, "Cannot insert 'null' as a List.");
		List<BaseIndexQuery> queries = new ArrayList<>();
		LOGGER.debug("{} - start ",Thread.currentThread().getStackTrace()[1].getMethodName());
		for (S s : entities) {
			queries.add(createIndexQuery(s));
		}
		LOGGER.debug("{} - createIndexQuery end ",Thread.currentThread().getStackTrace()[1].getMethodName());
		elasticsearchOperations.bulkIndexAsync(queries);
		return entities;
	}
	//done
	@Override
	public boolean existsById(ID id) {
		return findById(id).isPresent();
	}
	//done
	@Override
	public Iterable<T> search(QueryBuilder queryBuilder) {
		BaseSearchQuery query = new BaseNativeSearchQueryBuilder().withQuery(queryBuilder).build();
		int count = (int) elasticsearchOperations.count(query, getEntityClass());
		if (count == 0) {
			return new PageImpl<>(Collections.<T> emptyList());
		}
		query.setPageable(PageRequest.of(0, count));
		return elasticsearchOperations.queryForPage(query, getEntityClass());
	}
	//done
	@Override
	public Page<T> search(QueryBuilder queryBuilder, Pageable pageable) {
		BaseSearchQuery searchQuery = new BaseNativeSearchQueryBuilder().withQuery(queryBuilder).withPageable(pageable).build();
		return elasticsearchOperations.queryForPage(searchQuery, getEntityClass());
	}
	//done
	@Override
	public Page<T> search(BaseSearchQuery query) {
		return elasticsearchOperations.queryForPage(query, getEntityClass());
	}

	//done
	@Override
	public void deleteById(ID id) {
		Assert.notNull(id, "Cannot delete entity with id 'null'.");
		elasticsearchOperations.delete(entityInformation.getIndexName(), entityInformation.getType(),stringIdRepresentation(id));
	}
	//done
	@Override
	public void delete(T entity) {
		Assert.notNull(entity, "Cannot delete 'null' entity.");
		deleteById(extractIdFromBean(entity));
	}
	//done
	@Override
	public void deleteAll(Iterable<? extends T> entities) {
		Assert.notNull(entities, "Cannot delete 'null' list.");
		for (T entity : entities) {
			delete(entity);
		}
	}
	//done
	@Override
	public void deleteAll() {
		BaseDeleteQuery deleteQuery = new BaseDeleteQuery();
		deleteQuery.setQuery(matchAllQuery());
		elasticsearchOperations.delete(deleteQuery, getEntityClass());
	}


	private BaseIndexQuery createIndexQuery(T entity) {
		BaseIndexQuery query = new BaseIndexQuery();
		query.setObject(entity);
		query.setId(stringIdRepresentation(extractIdFromBean(entity)));
		query.setVersion(extractVersionFromBean(entity));
		query.setParentId(extractParentIdFromBean(entity));
		return query;
	}

	@SuppressWarnings("unchecked")
	private Class<T> resolveReturnedClassFromGenericType() {
		ParameterizedType parameterizedType = resolveReturnedClassFromGenericType(getClass());
		return (Class<T>) parameterizedType.getActualTypeArguments()[0];
	}

	private ParameterizedType resolveReturnedClassFromGenericType(Class<?> clazz) {
		Object genericSuperclass = clazz.getGenericSuperclass();
		if (genericSuperclass instanceof ParameterizedType) {
			ParameterizedType parameterizedType = (ParameterizedType) genericSuperclass;
			Type rawtype = parameterizedType.getRawType();
			if (BaseSimpleElasticsearchRepository.class.equals(rawtype)) {
				return parameterizedType;
			}
		}
		return resolveReturnedClassFromGenericType(clazz.getSuperclass());
	}

	@Override
	public Class<T> getEntityClass() {
		if (!isEntityClassSet()) {
			try {
				this.entityClass = resolveReturnedClassFromGenericType();
			} catch (Exception e) {
//				throw new InvalidDataAccessApiUsageException("Unable to resolve EntityClass. Please use according setter!", e);
				throw new RuntimeException("Unable to resolve EntityClass. Please use according setter!", e);
			}
		}
		return entityClass;
	}

	private boolean isEntityClassSet() {
		return entityClass != null;
	}

	public final void setEntityClass(Class<T> entityClass) {
		Assert.notNull(entityClass, "EntityClass must not be null.");
		this.entityClass = entityClass;
	}

	public final void setElasticsearchOperations(BaseElasticsearchOperations elasticsearchOperations) {
		Assert.notNull(elasticsearchOperations, "ElasticsearchOperations must not be null.");
		this.elasticsearchOperations = elasticsearchOperations;
	}

	protected ID extractIdFromBean(T entity) {
		return entityInformation.getId(entity);
	}

	private List<String> stringIdsRepresentation(Iterable<ID> ids) {
		Assert.notNull(ids, "ids can't be null.");
		List<String> stringIds = new ArrayList<>();
		for (ID id : ids) {
			stringIds.add(stringIdRepresentation(id));
		}
		return stringIds;
	}

	protected abstract String stringIdRepresentation(ID id);

	private Long extractVersionFromBean(T entity) {
		return entityInformation.getVersion(entity);
	}

	private String extractParentIdFromBean(T entity) {
		return entityInformation.getParentId(entity);
	}
}
